package day03;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 状态变量
 * @Author: ZYX
 * @Date: 2022/2/10 13:51
 * @Version: 1.0
 */
public class Demo05 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new SourceFunction<Integer>() {
            private boolean running = true;
            private Random random = new Random();

            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                while (running){
                    sourceContext.collect(random.nextInt(10));
                    TimeUnit.SECONDS.sleep(1L);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        }).keyBy(obj -> true)
                .process(new KeyedProcessFunction<Boolean, Integer, Double>() {
                    // 声明一个状态变量作为累加器
                    // 状态变量的可见范围是当前key
                    // 状态变量是单例只能被实例化一次
                    private ValueState<Tuple2<Integer,Integer>> valueState;
                    // 保存定时器时间戳
                    private ValueState<Long> timers;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        // 实例化变量
                        valueState = getRuntimeContext().getState(
                                // 状态描述符
                                new ValueStateDescriptor<Tuple2<Integer,Integer>>("sum-count", Types.TUPLE(Types.INT,Types.INT))
                        );

                        timers = getRuntimeContext().getState(
                                // 定时器
                                new ValueStateDescriptor<Long>("timer",Types.LONG)
                        );
                    }

                    @Override
                    public void processElement(Integer value, Context context, Collector<Double> out) throws Exception {
                        // 当第一条数据到来时，状态变量的值为null
                        // 使用.value()方法读取状态变量的值，使用.update()方法更新状态变量的值
                        if (valueState.value() == null) {
                            valueState.update(Tuple2.of(value,1));
                        }else {
                            Tuple2<Integer,Integer> tmp = valueState.value();
                            valueState.update(Tuple2.of(tmp.f0+value,tmp.f1+1));
                        }

                        if (timers.value() == null) {
                            long timerSeclater = context.timerService().currentProcessingTime() + 10 * 1000L;
                            context.timerService().registerProcessingTimeTimer(timerSeclater);
                            timers.update(timerSeclater);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Double> out) throws Exception {
                        super.onTimer(timestamp, ctx, out);
                        if (valueState.value() != null) {
                            System.out.println("f0:"+valueState.value().f0+"---f1:"+valueState.value().f1);
                            out.collect((double) valueState.value().f0 / valueState.value().f1);
                            timers.clear();
                        }
                    }
                }).print();

        env.execute();
    }

}
